package chapter10

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}

/**
 * author: 余辉
 * blog: https://blog.csdn.net/silentwolfyh
 * descriptions:
 * date: 2024 - 09 - 02 2:18 下午
 *
 *  有些场景下，逻辑不太方便用sql去实现，可能需要将dataframe退化成RDD来计算
 *  示例需求：  求每种性别的成绩总和
 */
object RDDToDF01 {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()
    // val schema = new StructType(Array(StructField("id",IntegerType),StructField("name",StringType)))

    // val schema = new StructType((StructField("id",IntegerType):: StructField("name",StringType) :: Nil).toArray)

    val schema = new StructType()
      .add("id", IntegerType)
      .add("name", StringType)
      .add("age", IntegerType)
      .add("sex", StringType)
      .add("city", StringType)
      .add("score", DoubleType)

    val df = spark.read.schema(schema).option("header", true).csv("BookData/input/10stu2.csv")

    // 可以直接在dataframe上用map等rdd算子
    // 框架会把算子返回的结果RDD 再转回dataset，需要一个能对RDD[T]进行解析的Encoder[T]才行
    // 好在大部分T类型都可以有隐式的Encoder来支持
    import spark.implicits._
    val ds2: Dataset[(Int, String)] = df.map(row => {
      val id = row.getAs[Int]("id")
      val name = row.getAs[String]("name")
      (id, name)
    })


    // dataframe中取出rdd后，就是一个RDD[Row]
    val rd: RDD[Row] = df.rdd
    // 从Row中取数据，就可以变成任意你想要的类型
    val rdd2: RDD[(Int, String, Int, String, String, Double)] = rd.map(row => {

      // dataframe是一种弱类型结构（在编译时无法检查类型，因为数据被装在了一个array[any]中）
      // val id = row.getDouble(1) // 如果类型取错，编译时是无法检查的，运行时才会报错

      // 可以根据字段的脚标去取
      val id: Int = row.getInt(0)
      val name: String = row.getString(1)
      val age: Int = row.getAs[Int](2)

      // 可以根据字段名称去取
      val sex: String = row.getAs[String]("sex")
      val city: String = row.getAs[String]("city")
      val score: Double = row.getAs[Double]("score")

      (id, name, age, sex, city, score)
    })


    /**
     * 用模式匹配从Row中抽取数据
     * 效果跟上面的方法是一样的，但是更简洁！
     */
    val rdd22 = rd.map({
      case Row(id: Int, name: String, age: Int, sex: String, city: String, score: Double) => {
        (id, name, age, sex, city, score)
      }
    })

    // 后续就跟dataframe没关系了，跟以前的rdd是一样的了
    val res: RDD[(String, Double)] = rdd22.groupBy(tp => tp._4).mapValues(iter => {
      iter.map(_._6).sum
    })

    res.foreach(println)

    spark.close()
  }
}